{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "56ea28e3-c67e-4177-a6f3-39395c221e3e",
   "metadata": {},
   "source": [
    "## A03 Move Data to the Exploitation Zone\n",
    "\n",
    "This code will perform data processing and cleaning"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e0228301-8b03-470f-82a1-05a8989a0d03",
   "metadata": {},
   "source": [
    "### 1. init spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "c048f9f9-e40c-466a-87db-ceb6c148601a",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c16a71dc-69df-40ba-b06a-f66d4d319fbb",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"price_opendata\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f63fe62e-8424-46ea-9e7b-0a0520157f58",
   "metadata": {},
   "source": [
    "### 2. read the parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "1ef8ec7a-d7b8-43ac-9af5-0b4f5ca0d659",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+--------------+----------------+------+--------+----------+------------+----------+------------+----+\n",
      "|_id|district_id| district_name|     neigh_name |Amount|PerMeter|diffAmount|diffPerMeter|usedAmount|usedPerMeter|year|\n",
      "+---+-----------+--------------+----------------+------+--------+----------+------------+----------+------------+----+\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 176.4|  2331.7|     237.1|      2987.8|     170.7|      2270.0|2013|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 201.1|  2546.2|     213.7|      2971.7|     200.5|      2528.9|2014|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 225.7|  3052.2|     302.9|      4301.5|     223.2|      3010.8|2015|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 222.8|  2991.0|      NULL|        NULL|     222.9|      2994.9|2016|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 272.5|  3552.0|     385.6|      4065.2|     265.1|      3518.3|2017|\n",
      "| 33|          7|Horta-Guinardó|        Can Baró| 192.6|  1823.4|      NULL|        NULL|     192.6|      1823.4|2013|\n",
      "| 33|          7|Horta-Guinardó|        Can Baró| 178.3|  2118.8|      NULL|        NULL|     178.3|      2118.8|2014|\n",
      "| 33|          7|Horta-Guinardó|        Can Baró| 129.1|  2101.1|      NULL|        NULL|     129.1|      2101.1|2015|\n",
      "| 33|          7|Horta-Guinardó|        Can Baró| 164.9|  2157.1|      NULL|        NULL|     164.9|      2157.1|2016|\n",
      "| 33|          7|Horta-Guinardó|        Can Baró| 221.3|  2423.7|      NULL|        NULL|     221.3|      2423.7|2017|\n",
      "+---+-----------+--------------+----------------+------+--------+----------+------------+----------+------------+----+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "file_path = \"./Formatted Zone/price_opendata.parquet\"\n",
    "\n",
    "df = spark.read.parquet(file_path)\n",
    "\n",
    "df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "cd3c7c04-9a13-44d4-a2cd-3cb8d57666e8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: long (nullable = true)\n",
      " |-- district_id: long (nullable = true)\n",
      " |-- district_name: string (nullable = true)\n",
      " |-- neigh_name : string (nullable = true)\n",
      " |-- Amount: double (nullable = true)\n",
      " |-- PerMeter: double (nullable = true)\n",
      " |-- diffAmount: double (nullable = true)\n",
      " |-- diffPerMeter: double (nullable = true)\n",
      " |-- usedAmount: double (nullable = true)\n",
      " |-- usedPerMeter: double (nullable = true)\n",
      " |-- year: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f138063e-d834-4216-8a65-53263fc6ba3f",
   "metadata": {},
   "source": [
    "### 3. describe the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "949bf219-6b0d-49a4-ac74-6303165869e9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+-----------------+-------------------+---------------+------------------+------------------+------------------+------------------+------------------+------------------+------------------+\n",
      "|summary|               _id|      district_id|      district_name|    neigh_name |            Amount|          PerMeter|        diffAmount|      diffPerMeter|        usedAmount|      usedPerMeter|              year|\n",
      "+-------+------------------+-----------------+-------------------+---------------+------------------+------------------+------------------+------------------+------------------+------------------+------------------+\n",
      "|  count|               359|              359|                359|            359|               359|               359|               210|               211|               357|               357|               359|\n",
      "|   mean| 36.66295264623955|6.245125348189415|               NULL|           NULL|239.66350974930364| 2732.004735376044|296.48952380952386| 3277.741232227488|237.67450980392167| 2708.732212885151|2015.0055710306406|\n",
      "| stddev|21.052887464363092| 2.79785880288695|               NULL|           NULL|150.50403725145722|1073.8223673650552| 200.0191557447534|1295.7169122247115|  151.920389393897|1076.3737064578443| 1.418147404637931|\n",
      "|    min|                 1|                1|       Ciutat Vella|  Baró de Viver|              48.5|             342.6|              73.0|             342.6|              48.5|             612.6|              2013|\n",
      "|    max|                73|               10|Sarrià-Sant Gervasi|les Tres Torres|             890.0|            6951.3|            1642.0|            7786.2|             890.0|            7020.2|              2017|\n",
      "+-------+------------------+-----------------+-------------------+---------------+------------------+------------------+------------------+------------------+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a1613d0b-48e9-4af6-adb2-67c59f379228",
   "metadata": {},
   "source": [
    "### 4. Remove spaces from column names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "582d2d12-2fbe-4b43-b12f-7bbf8139c814",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumnRenamed(\"neigh_name \", \"neigh_name\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d63b307-4650-4690-967e-dd953694d427",
   "metadata": {},
   "source": [
    "### 5. check the null fields"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "82cbf526-d3c6-43fb-98c4-98afaf5c4da7",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import when, col, sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "f69ce964-d7e4-42c3-9838-4e9bc3397c30",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "|_id|district_id|district_name|neigh_name|Amount|PerMeter|diffAmount|diffPerMeter|usedAmount|usedPerMeter|year|\n",
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "|  0|          0|            0|         0|     0|       0|       149|         148|         2|           2|   0|\n",
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "missing_values_df = df.select(\n",
    "    [sum(when(col(column).isNull(), 1).otherwise(0)).alias(column) for column in df.columns]\n",
    ")\n",
    "\n",
    "missing_values_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc698566-cf0a-4fac-9158-2dfa68158a86",
   "metadata": {},
   "source": [
    "### 6. Columns with missing values filled using the mean value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "256951f9-79f1-436d-9e44-d26691a252ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "columns_to_fill = [\"diffAmount\", \"diffPerMeter\", \"usedAmount\", \"usedPerMeter\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "d0bd2ed1-0d09-43d8-8aa2-a7e3c7a9bd48",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, mean"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "cd5a93c1-bf89-442e-b7a2-f44e66cc1428",
   "metadata": {},
   "outputs": [],
   "source": [
    "# compute the mean values\n",
    "means = df.select([mean(col(c)).alias(c) for c in columns_to_fill]).first().asDict()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "41bc7abf-2d4d-494e-a34c-9d28646220ec",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'diffAmount': 296.48952380952386,\n",
       " 'diffPerMeter': 3277.741232227488,\n",
       " 'usedAmount': 237.67450980392167,\n",
       " 'usedPerMeter': 2708.732212885151}"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "means"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "3c8298d0-5128-4dbc-b9da-65f3ab91d719",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+--------------+----------------+------+--------+------------------+-----------------+----------+------------+----+\n",
      "|_id|district_id| district_name|      neigh_name|Amount|PerMeter|        diffAmount|     diffPerMeter|usedAmount|usedPerMeter|year|\n",
      "+---+-----------+--------------+----------------+------+--------+------------------+-----------------+----------+------------+----+\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 176.4|  2331.7|             237.1|           2987.8|     170.7|      2270.0|2013|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 201.1|  2546.2|             213.7|           2971.7|     200.5|      2528.9|2014|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 225.7|  3052.2|             302.9|           4301.5|     223.2|      3010.8|2015|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 222.8|  2991.0|296.48952380952386|3277.741232227488|     222.9|      2994.9|2016|\n",
      "| 32|          7|Horta-Guinardó|el Baix Guinardó| 272.5|  3552.0|             385.6|           4065.2|     265.1|      3518.3|2017|\n",
      "+---+-----------+--------------+----------------+------+--------+------------------+-----------------+----------+------------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df = df.fillna(means)\n",
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "6e9de0d3-7dff-4d93-af57-56831f0bbc19",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "|_id|district_id|district_name|neigh_name|Amount|PerMeter|diffAmount|diffPerMeter|usedAmount|usedPerMeter|year|\n",
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "|  0|          0|            0|         0|     0|       0|         0|           0|         0|           0|   0|\n",
      "+---+-----------+-------------+----------+------+--------+----------+------------+----------+------------+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "missing_values_df = df.select(\n",
    "    [sum(when(col(column).isNull(), 1).otherwise(0)).alias(column) for column in df.columns]\n",
    ")\n",
    "\n",
    "missing_values_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7afc5c63-6e8c-4a2c-9b6f-dc6e5546269a",
   "metadata": {},
   "source": [
    "### 7. save to parquet file and csv file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "6fac4817-5259-47fd-9cf0-e0fa6a7ab96b",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Exploitation Zone/price_opendata.parquet\"\n",
    "\n",
    "df.write.mode(\"overwrite\").parquet(output_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "6293c3fd-68c7-43ad-a1fe-550ecb9c27ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Exploitation Zone/price_opendata.csv\"\n",
    "\n",
    "df.write.mode(\"overwrite\").csv(output_path, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cdb90bfa-023e-44a8-b754-120a88d88be0",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
